package com.hmc.kafka.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * 调用生产者发送到缓冲区后回调方法
 */
public class CustomProducerCallback {
    public static void main(String[] args) throws InterruptedException {
        Properties properties = CustomProducer.getProperties();
        //配置自定义分区策略
//        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.hmc.kafka.producer.MyPartitioner");

        //创建生产者
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        //发送数据
        for (int i = 1 ; i<=50 ;i++) {
            producer.send(new ProducerRecord<String, String>("third", "hmc" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e == null) {
                        System.out.println("topic:\t"+recordMetadata.topic()+"\t\tpartition:\t"+recordMetadata.partition());
                    }
                }
            });
            Thread.sleep(2);
            producer.send(new ProducerRecord<String, String>("second", "test" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e == null) {
                        System.out.println("topic:\t"+recordMetadata.topic()+"\t\tpartition:\t"+recordMetadata.partition());
                    }
                }
            });
        }
        //关闭资源
        producer.close();
    }
}
